Overview

1. Fixes issue AdminClient is stuck at CONNECT when using OAUTHBEARER with SASL_SSL protocol #1713 by delegating OAuth token refresh events to be handled by the background thread.
2. This has been implemented for Admin, Producer and Consumer clients to make behaviour consistent with the Java client.

Note that, poll() will not trigger the OAuth refresh callbacks anymore.

Problem

When providing sasl.mechanisms as OAUTHBEARER and a oauth_cb in the conf, users were unable to make API calls like list_topics(), create_topics() unless poll was called.

>>> from confluent_kafka.admin import AdminClient, NewTopic
>>> from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
>>> import socket

>>> def oauth_cb(oauth_config):
...     auth_token, expiry_ms = MSKAuthTokenProvider.generate_auth_token("region")
...     return auth_token, expiry_ms/1000

>>> config = {
...     'bootstrap.servers': 'xxx-1.kafka.us-west-2.amazonaws.com:9098,xxx-2.kafka.us-west-2.amazonaws.com:9098',
...     'security.protocol': 'SASL_SSL',
...     'sasl.mechanism': 'OAUTHBEARER',
...     'oauth_cb': oauth_cb,
... }

>>> admin = AdminClient(config)
>>> admin.list_topics(timeout=2)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/venv3/lib/python3.8/site-packages/confluent_kafka/admin/__init__.py", line 603, in list_topics
    return super(AdminClient, self).list_topics(*args, **kwargs)
cimpl.KafkaException: KafkaError{code=_TRANSPORT,val=-195,str="Failed to get metadata: Local: Broker transport failure"}

(Snippet Source: #1713 (comment))

This happens because OAuth token refresh events get pushed to the main queue which is listened by poll(). These events are responsible for triggering the configured OAuth callback function and setting the token in the request. If poll() is not called before list_topics(), the token would not be set in the request, causing the request to fail.

Solution

Changes have been made to handle these refresh events in the background thread using the following librdkafka method calls:

1. rd_kafka_conf_enable_sasl_queue(conf, 1) - ensures all refresh events get pushed to the SASL queue instead of main queue

2. rd_kafka_sasl_background_callbacks_enable(self->rk) - ensures all events in SASL queue are forwarded to the background queue.

By using these 2 method calls, we have been able to ensure that OAuth callbacks get triggered from the background thread instead of poll()

Checklist

• Contains customer facing changes? Including API/behavior changes
• Did you add sufficient unit test and/or integration test coverage for this PR?